package cn.zwx.canal.tools.listener;

import cn.hutool.core.collection.CollectionUtil;
import cn.zwx.canal.tools.event.AbstractCanalEvent;
import cn.zwx.canal.tools.kafka.KafKaRegisterProducer;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;

import java.util.List;
import java.util.Map;


/**
 * @description: canal监听基类
 * 只监听我们自己封装的事件
 * @projectName:learn
 * @see:cn.zwx.canal.tools.listener
 * @author:zhangwenxue
 * @createTime:2020/12/19 12:03
 * @version:1.0
 */
public abstract class AbstractCanalListener <EVENT extends AbstractCanalEvent> implements ApplicationListener<EVENT> {

    private Logger logger = LoggerFactory.getLogger(AbstractCanalListener.class);
    @Autowired
    private KafKaRegisterProducer kafKaRegisterProducer;

    @Override
    public void onApplicationEvent(EVENT event) {
        // 获取监听对象 
        CanalEntry.Entry entry = event.getEntry();
        // 获取数据库名称
        String database = entry.getHeader().getSchemaName();
        // 获取表名称 
        String tableName = entry.getHeader().getTableName();
        // 获取变动元素
        CanalEntry.RowChange rowChange;
        try {
            rowChange =  CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        }catch (InvalidProtocolBufferException e){
            logger.error("canalEntry_parser_error,根据CanalEntry获取RowChange失败！", e);
            return;
        }
        
        // 具体处理交子类 
        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
        if (CollectionUtil.isEmpty(rowDataList)){
            logger.error("未获取到变动数据的rowData");
            return;
        }
        rowDataList.stream().forEach(rowData -> {
            this.doChange(database,tableName,rowData);
        });

    }

    /**
     * @description 对应具体业务处理逻辑
     * @param database 数据库名称
     * @param tableName 表名称
     * @param rowData  变动数据
     * @return void
     * @author zhangwenxue
     * @createTime 2020/12/19 12:15
     **/
    protected abstract void doChange(String database, String tableName, CanalEntry.RowData rowData);

    /**
     * @description 消息发送处理类
     * @return void
     * @author zhangwenxue
     * @createTime 2021/1/12 15:28
     **/
    protected void sendMessage(Map<String,Object> req){
        kafKaRegisterProducer.sendMessage(JSON.toJSONString(req));
    }




}

